import numpy as np
import dask.dataframe as dd

class TechnicalIndicator:
    """技术指标计算工具集"""
    
    @staticmethod
    def moving_average(data: dd.DataFrame, window: int = 20) -> dd.Series:
        """滚动均线（支持dask并行计算）"""
        return data['close'].rolling(window).mean()
    
    @staticmethod
    def rsi(data: dd.DataFrame, period: int = 14) -> dd.Series:
        """相对强弱指数"""
        delta = data['close'].diff()
        gain = delta.where(delta > 0, 0)
        loss = -delta.where(delta < 0, 0)
        
        avg_gain = gain.rolling(period).mean()
        avg_loss = loss.rolling(period).mean()
        
        rs = avg_gain / avg_loss
        return 100 - (100 / (1 + rs))
    
    @staticmethod
    def macd(data: dd.DataFrame, 
            fast: int = 12, 
            slow: int = 26, 
            signal: int = 9) -> dd.DataFrame:
        """MACD指标（返回DIF, DEA, MACD三列）"""
        ema_fast = data['close'].ewm(span=fast).mean()
        ema_slow = data['close'].ewm(span=slow).mean()
        dif = ema_fast - ema_slow
        dea = dif.ewm(span=signal).mean()
        macd = (dif - dea) * 2
        return dd.concat([dif, dea, macd], axis=1, 
                        columns=['DIF', 'DEA', 'MACD'])